#include <atomic>
#include <chrono>
#include <cstddef>
#include <cstdint>
#include <deque>
#include <iomanip>
#include <iostream>
#include <mutex>
#include <string>
#include <vector>
#if defined(__APPLE__) || defined(__linux__)
#include <cstdlib>
#include <unistd.h>
#endif

#include "asycipc/api.hpp"
#include "asycipc/frame.hpp"
#include "asycipc/named_pipe.hpp"

namespace {

using namespace asycipc;

std::vector<uint8_t> ToBytes(const std::string& text) {
  return std::vector<uint8_t>(text.begin(), text.end());
}

std::string ToString(const std::vector<uint8_t>& data) {
  return std::string(data.begin(), data.end());
}

enum class DemoPhase { kServerInitiated, kClientInitiated, kLargeTransfer };

void SendFrames(PipeHandle& handle, const std::vector<Frame>& frames,
                std::mutex& write_mutex) {
  for (const auto& frame : frames) {
    auto bytes = frame.Serialize();
    std::lock_guard<std::mutex> lock(write_mutex);
    handle.Write(bytes);
  }
}

void SendMaintenance(PipeServer& server, Channel& channel, PipeHandle& handle,
                     std::mutex& write_mutex) {
  auto frames = server.CollectMaintenanceFrames(channel);
  SendFrames(handle, frames, write_mutex);
}

std::vector<uint8_t> MakeLargePayload(std::size_t size, uint8_t seed) {
  std::vector<uint8_t> data(size);
  for (std::size_t i = 0; i < size; ++i) {
    data[i] = static_cast<uint8_t>(seed + static_cast<uint8_t>(i));
  }
  return data;
}

constexpr std::size_t kFrameHeaderSize = 24;
constexpr std::size_t kFrameTrailerSize = 4;

std::size_t NextFrameSize(const std::vector<uint8_t>& buffer) {
  if (buffer.size() < kFrameHeaderSize) {
    return 0;
  }
  const uint16_t magic = static_cast<uint16_t>(static_cast<uint16_t>(buffer[0]) << 8) |
                         static_cast<uint16_t>(buffer[1]);
  if (magic != kFrameMagic) {
    throw std::runtime_error("invalid frame magic in stream");
  }
  if (buffer[2] != kProtocolVersion) {
    throw std::runtime_error("unsupported frame version in stream");
  }
  const uint32_t payload_length =
      (static_cast<uint32_t>(buffer[14]) << 24) |
      (static_cast<uint32_t>(buffer[15]) << 16) |
      (static_cast<uint32_t>(buffer[16]) << 8) |
      static_cast<uint32_t>(buffer[17]);
  const uint16_t ext_length = static_cast<uint16_t>((static_cast<uint16_t>(buffer[22]) << 8) |
                                                    static_cast<uint16_t>(buffer[23]));

  const std::size_t total = kFrameHeaderSize + static_cast<std::size_t>(ext_length) +
                            static_cast<std::size_t>(payload_length) +
                            kFrameTrailerSize;
  if (buffer.size() < total) {
    return 0;
  }
  return total;
}

bool DrainServerBuffer(PipeServer& server, Channel& channel, PipeHandle& handle,
                       std::mutex& write_mutex, std::vector<uint8_t>& buffer) {
  bool processed = false;
  while (true) {
    const std::size_t frame_size = NextFrameSize(buffer);
    if (frame_size == 0 || buffer.size() < frame_size) {
      break;
    }
    std::vector<uint8_t> frame_bytes(buffer.begin(),
                                     buffer.begin() + static_cast<std::ptrdiff_t>(frame_size));
    buffer.erase(buffer.begin(),
                 buffer.begin() + static_cast<std::ptrdiff_t>(frame_size));

    auto outbound = server.ProcessInbound(channel, frame_bytes);
    SendFrames(handle, outbound, write_mutex);
    SendMaintenance(server, channel, handle, write_mutex);
    processed = true;
  }
  return processed;
}

bool PumpClient(PipeServer& server, Channel& channel, PipeHandle& handle,
                std::mutex& write_mutex, std::vector<uint8_t>& read_buffer) {
  while (true) {
    if (DrainServerBuffer(server, channel, handle, write_mutex, read_buffer)) {
      return true;
    }

    std::vector<uint8_t> payload;
    try {
      payload = handle.Read();
    } catch (const std::exception& ex) {
      std::cerr << "[server] read failed: " << ex.what() << std::endl;
      return false;
    }
    if (payload.empty()) {
      return false;
    }

    read_buffer.insert(read_buffer.end(), payload.begin(), payload.end());
  }
}

}  // namespace

#if defined(__APPLE__)
std::string DefaultEndpoint() {
  if (const char* override_dir = std::getenv("ASYIPC_RUNTIME_DIR")) {
    if (*override_dir) {
      return std::string(override_dir) + "/demo";
    }
  }
  if (geteuid() == 0) {
    return "/var/run/asycipc/demo";
  }
  if (const char* xdg = std::getenv("XDG_RUNTIME_DIR")) {
    if (*xdg) {
      return std::string(xdg) + "/asycipc/demo";
    }
  }
  if (const char* tmp = std::getenv("TMPDIR")) {
    if (*tmp) {
      return std::string(tmp) + "/asycipc/demo";
    }
  }
  return std::string("/tmp/asycipc/demo");
}
#elif defined(__linux__)
std::string DefaultEndpoint() {
  if (const char* override_dir = std::getenv("ASYIPC_RUNTIME_DIR")) {
    if (*override_dir) {
      return std::string(override_dir) + "/demo";
    }
  }
  if (geteuid() == 0) {
    return "/run/asycipc/demo";
  }
  if (const char* xdg = std::getenv("XDG_RUNTIME_DIR")) {
    if (*xdg) {
      return std::string(xdg) + "/asycipc/demo";
    }
  }
  return "/run/user/" + std::to_string(geteuid()) + "/asycipc/demo";
}
#else
std::string DefaultEndpoint() {
  return R"(\\.\pipe\asycipc_demo)";
}
#endif

int main(int argc, char** argv) {
  std::string pipe_endpoint = DefaultEndpoint();
  if (argc > 1) {
    pipe_endpoint = argv[1];
  }

  std::deque<std::vector<uint8_t>> inbox;
  std::atomic<DemoPhase> phase{DemoPhase::kServerInitiated};

  auto server_builder = PipeServer::Builder()
                            .Endpoint("demo")
                            .AllowApp("demo-client")
                            .OnMessage([&](Channel& channel, const Frame& frame) {
                              inbox.push_back(frame.payload);

                              if (phase.load() == DemoPhase::kClientInitiated) {
                                std::string text = ToString(frame.payload);
                                const std::string kPrefix = "client ping ";
                                if (text.rfind(kPrefix, 0) == 0) {
                                  std::string suffix = text.substr(kPrefix.size());
                                  std::string reply = "server pong " + suffix;
                                  return channel.MakeDataFrames(ToBytes(reply));
                                }
                              }
                              return std::vector<Frame>{};
                            })
                            .OnError([](const std::string& msg) {
                              std::cerr << "[server] error: " << msg << std::endl;
                            });
#if defined(__APPLE__)
  server_builder.ReliabilityEnabled(false);
#endif
  auto server = server_builder.Build();
  server.Start();

  auto listener = CreateNamedPipeListener({pipe_endpoint});
  std::cout << "[server] listening on " << pipe_endpoint << std::endl;

  auto handle = listener->Accept();
  std::cout << "[server] client connected" << std::endl;

  const auto handshake_bytes = handle->Read();
  std::cout << "[server] handshake request: " << ToString(handshake_bytes) << std::endl;

  SessionContext session;
  HandshakeRequest request;
  request.app_id = "demo-client";
  request.capabilities = {"demo"};
  HandshakeResponse response = server.Accept(request, session);
  (void)response;
  handle->Write(ToBytes("READY\n"));

  auto channel = server.CreateChannel(std::move(session));
  channel->SetTimeProvider([] { return std::chrono::steady_clock::now(); });

  std::mutex write_mutex;
  std::vector<uint8_t> read_buffer;

  auto wait_for_message = [&]() {
    while (inbox.empty()) {
      if (!PumpClient(server, *channel, *handle, write_mutex, read_buffer)) {
        throw std::runtime_error("connection closed");
      }
    }
    auto data = std::move(inbox.front());
    inbox.pop_front();
    return data;
  };

  auto send_payload = [&](const std::vector<uint8_t>& data) {
    auto frames = channel->MakeDataFrames(data);
    SendFrames(*handle, frames, write_mutex);
    SendMaintenance(server, *channel, *handle, write_mutex);
  };

  try {
    std::cout << "[server] phase 1: server -> client pings" << std::endl;
    for (int i = 0; i < 100; ++i) {
      std::string ping = "server ping " + std::to_string(i);
      send_payload(ToBytes(ping));

      auto reply = wait_for_message();
      std::string reply_text(reply.begin(), reply.end());
      std::cout << "  [server] received: " << reply_text << std::endl;
    }

    phase.store(DemoPhase::kClientInitiated);
    std::cout << "[server] phase 2: client -> server pings" << std::endl;
    for (int i = 0; i < 100; ++i) {
      auto message = wait_for_message();
      std::string text(message.begin(), message.end());
      std::cout << "  [server] received: " << text << std::endl;
    }

    phase.store(DemoPhase::kLargeTransfer);
    std::cout << "[server] phase 3: 4MB payload round-trip" << std::endl;
    const std::size_t payload_size = 4 * 1024 * 1024;
    for (int i = 0; i < 100; ++i) {
      auto payload = MakeLargePayload(payload_size, static_cast<uint8_t>(i));
      send_payload(payload);

      auto reply = wait_for_message();
      if (reply.size() != payload.size()) {
        std::cerr << "  [server] malformed reply size=" << reply.size() << std::endl;
        continue;
      }
      bool valid = true;
      for (std::size_t idx = 0; idx < reply.size(); ++idx) {
        if (reply[idx] != static_cast<uint8_t>(~payload[idx])) {
          valid = false;
          break;
        }
      }
      std::cout << "  [server] chunk " << std::setw(3) << i
                << (valid ? " OK" : " MISMATCH") << std::endl;
    }
  } catch (const std::exception& ex) {
    std::cerr << "[server] termination: " << ex.what() << std::endl;
  }

  handle.reset();

  std::cout << "[server] demo completed" << std::endl;
  return 0;
}
